package bch.kafka.integrate.controller;

import bch.kafka.integrate.stream.UserRiskStream;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.function.Function;

@RestController
@RequestMapping("/api")
public class Api {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @RequestMapping("/send")
    public String send(@RequestParam("topic") String topic, @RequestParam("message") String message) {
        KafkaAdmin kafkaAdmin = kafkaTemplate.getKafkaAdmin();
        String defaultTopic = kafkaTemplate.getDefaultTopic();
        RecordMessageConverter messageConverter = kafkaTemplate.getMessageConverter();
        ProducerFactory<String, String> producerFactory = kafkaTemplate.getProducerFactory();
        Function<ProducerRecord<?, ?>, Map<String, String>> micrometerTagsProvider = kafkaTemplate.getMicrometerTagsProvider();
        kafkaTemplate.send(topic, message);
        return "Hello World!";
    }

    @RequestMapping("/send2")
    public String send2() {
        String rep = """
                {"userId":"u123","event":"login_fail","timestamp":1695555555555}
                """;
        for (int i = 0; i < 5; i++) {
            kafkaTemplate.send(UserRiskStream.USER_EVENTS_TOPIC, rep);
        }
        return "Hello World!";
    }
}
